package com.hmdp.service.impl;

import cn.hutool.core.bean.BeanUtil;
import com.hmdp.dto.Result;
import com.hmdp.dto.UserDTO;
import com.hmdp.entity.SeckillVoucher;
import com.hmdp.entity.VoucherOrder;
import com.hmdp.mapper.SeckillVoucherMapper;
import com.hmdp.mapper.VoucherOrderMapper;
import com.hmdp.service.ISeckillVoucherService;
import com.hmdp.service.IVoucherOrderService;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.hmdp.utils.RedisIdWorker;
import com.hmdp.utils.SimpleRedisLock;
import com.hmdp.utils.UserHolder;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.aop.framework.AopContext;
import org.springframework.core.io.ClassPathResource;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.time.Duration;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * <p>
 *  服务实现类
 * </p>
 *
 * @author 虎哥
 * @since 2021-12-22
 */
@Service
@Slf4j
public class VoucherOrderServiceImpl extends ServiceImpl<VoucherOrderMapper, VoucherOrder> implements IVoucherOrderService {

    @Resource
    ISeckillVoucherService seckillVoucherService;

    @Resource
    SeckillVoucherMapper seckillVoucherMapper;

    @Resource
    RedisIdWorker redisIdWorker;

    @Resource
    IVoucherOrderService voucherOrderService;

    @Resource
    StringRedisTemplate stringRedisTemplate;

    @Resource
    RedissonClient redissonClient;

    // 扩大代理作用域,线程池才能拿到
    private IVoucherOrderService proxy;

    private static final DefaultRedisScript<Long> SECKILL_SCRIPT;
    // 初始化脚本
    static {
        SECKILL_SCRIPT = new DefaultRedisScript();
        SECKILL_SCRIPT.setLocation(new ClassPathResource("seckill.lua"));
        SECKILL_SCRIPT.setResultType(Long.class);
    }

    // 初始化阻塞队列
    // private BlockingQueue<VoucherOrder> orderTasks = new ArrayBlockingQueue<>(1024 * 1024);
    // 初始化线程池
    private static final ExecutorService SKILL_ORDER_EXECUTOR = Executors.newSingleThreadExecutor();
    // 当前类初始化完毕后会执行 @PostConstruct 标注的方法
    @PostConstruct
    private void init() {
        // 执行线程池
        SKILL_ORDER_EXECUTOR.submit(new VoucherOrderHandler());
    }

    // 开启线程
    private class VoucherOrderHandler implements Runnable {
        @Override
        public void run() {
            while (true) {
                try {
                    List<MapRecord<String, Object, Object>> list = stringRedisTemplate.opsForStream().read(
                            // g1消费者组 c1消费者
                            Consumer.from("g1", "c1"),
                            // 最大获取消息数为1条 没有消息则阻塞2秒
                            StreamReadOptions.empty().count(1).block(Duration.ofSeconds(2)),
                            // 通过g1消费者组的c1消费者 读取队列名称为stream.orders的消息  lastConsumed是">"，表示消费未消费的消息
                            StreamOffset.create("stream.orders", ReadOffset.lastConsumed())
                    );
                    // 判断获取消息是否成功 没有消息则继续循环
                    if (list == null || list.isEmpty()) {
                        continue;
                    }
                    // 由于上方只会获取一条消息，存在消息我们则只需获取下标0的消息即可
                    MapRecord<String, Object, Object> record = list.get(0);
                    // 获取消息中的value值，如下行注释 我们在脚本中存消息的时候是k1 v1 k2 v2键值对的形式存储的
                    // redis.call('xadd', 'stream.orders', '*', 'userId', userId, 'voucherId', voucherId, 'id', orderId)
                    Map<Object, Object> values = record.getValue();
                    // key value转换成订单对象
                    VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(values, new VoucherOrder(), true);
                    // 订单获取成功则下单
                    handleVoucherOrder(voucherOrder);
                    // 确认ACK 向g1组的stream.orders队列发送消息id进行ack确认
                    stringRedisTemplate.opsForStream().acknowledge("stream.orders", "g1", record.getId());

                } catch (Exception e) {
                    log.error("处理订单出现异常", e);
                    handlePendingList();
                }
            }
        }
        // 处理pending-list未消费完成的消息
        private void handlePendingList() {
            while (true) {
                try {
                    List<MapRecord<String, Object, Object>> list = stringRedisTemplate.opsForStream().read(
                            // g1消费者组 c1消费者
                            Consumer.from("g1", "c1"),
                            // 最大获取消息数为1条
                            StreamReadOptions.empty().count(1),
                            // 通过g1消费者组的c1消费者 读取队列名称为stream.orders的消息  "0"表示从pending-list获取未消费完成的消息
                            StreamOffset.create("stream.orders", ReadOffset.from("0"))
                    );
                    // 判断获取消息是否成功 没有消息则跳出pending-list循环
                    if (list == null || list.isEmpty()) {
                        break;
                    }
                    // 由于上方只会获取一条消息，存在消息我们则只需获取下标0的消息即可
                    MapRecord<String, Object, Object> record = list.get(0);
                    // 获取消息中的value值，如下行注释 我们在脚本中存消息的时候是k1 v1 k2 v2键值对的形式存储的
                    // redis.call('xadd', 'stream.orders', '*', 'userId', userId, 'voucherId', voucherId, 'id', orderId)
                    Map<Object, Object> values = record.getValue();
                    // key value转换成订单对象
                    VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(values, new VoucherOrder(), true);
                    // 订单获取成功则下单
                    handleVoucherOrder(voucherOrder);
                    // 确认ACK 向g1组的stream.orders队列发送消息id进行ack确认
                    stringRedisTemplate.opsForStream().acknowledge("stream.orders", "g1", record.getId());
                } catch (Exception e) {
                    log.error("处理pending-list订单出现异常", e);
                }
            }
        }
    }


    // // 开启线程
    // private class VoucherOrderHandler implements Runnable {
    //     @Override
    //     public void run() {
    //         while (true) {
    //             try {
    //                 // 从阻塞队列中不断取出订单 执行创建订单逻辑
    //                 VoucherOrder voucherOrder = orderTasks.take();
    //                 handleVoucherOrder(voucherOrder);
    //             } catch (Exception e) {
    //                 log.error("处理订单出现异常", e);
    //             }
    //         }
    //     }
    // }

    // 创建订单
    private void handleVoucherOrder(VoucherOrder voucherOrder) {
        // 这个方法会在线程中执行,直接从UserHolder中获取用户是获取不到的,只能从传过来的对象中拿
        Long userId = voucherOrder.getUserId();
        // -- 这里实际可以去掉判断重复下单的加锁解锁操作,因为在redis中已经进行了判断,这里是二次兜底防止redis出现意外
        RLock lock = redissonClient.getLock("lock:order:" + userId);
        boolean isLock = lock.tryLock();
        if (!isLock) {
            log.error("不允许重复下单");
            return;
        }
        try {
            proxy.createVoucherOrder(voucherOrder);
        } finally {
            lock.unlock();
        }
    }

    // 接口调的方法,执行lua脚本和订单存入阻塞队列
    @Override
    public Result seckillVoucher(Long voucherId) {
        // 获取用户id
        Long userId = UserHolder.getUser().getId();
        // 生成订单id
        long orderId = redisIdWorker.nextId("order");
        // 执行lua脚本
        Long execute = stringRedisTemplate.execute(SECKILL_SCRIPT, Collections.emptyList(), voucherId.toString(), userId.toString(), String.valueOf(orderId));
        int result = execute.intValue();
        // 判断返回结果
        // 不是0说明没有购买资格
        if (result != 0) {
            // 返回1说明库存不足 返回2说明已有该优惠券购买记录
            return Result.fail(result == 1 ? "库存不足" : "不能重复下单");
        }

        // 获取当前代理对象 用于阻塞队列使用
        proxy = (IVoucherOrderService) AopContext.currentProxy();
        return Result.ok(orderId);
    }

    // // 接口调的方法,执行lua脚本和订单存入阻塞队列
    // @Override
    // public Result seckillVoucher(Long voucherId) {
    //     // 获取用户id
    //     Long userId = UserHolder.getUser().getId();
    //     // 执行lua脚本
    //     Long execute = stringRedisTemplate.execute(SECKILL_SCRIPT, Collections.emptyList(), voucherId.toString(), userId.toString());
    //     int result = execute.intValue();
    //     // 判断返回结果
    //     // 不是0说明没有购买资格
    //     if (result != 0) {
    //         // 返回1说明库存不足 返回2说明已有该优惠券购买记录
    //         return Result.fail(result == 1 ? "库存不足" : "不能重复下单");
    //     }
    //     // 生成订单id
    //     long orderId = redisIdWorker.nextId("order");
    //     // 创建订单
    //     VoucherOrder voucherOrder = new VoucherOrder();
    //     voucherOrder.setId(orderId);
    //     voucherOrder.setUserId(userId);
    //     voucherOrder.setVoucherId(voucherId);
    //     // 存入阻塞队列
    //     orderTasks.add(voucherOrder);
    //     // 获取当前代理对象 用于阻塞队列使用
    //     proxy = (IVoucherOrderService) AopContext.currentProxy();
    //     return Result.ok(orderId);
    // }

    // @Override
    // public Result seckillVoucher(Long voucherId) {
    //     SeckillVoucher seckillVoucher = seckillVoucherService.getById(voucherId);
    //     if (seckillVoucher == null) {
    //         return Result.fail("该优惠券不存在！");
    //     }
    //     // 开始时间在当前时间之后
    //     if (seckillVoucher.getBeginTime().isAfter(LocalDateTime.now())) {
    //         return Result.fail("秒杀未开始！");
    //     }
    //     // 结束时间在当前时间之前
    //     if (seckillVoucher.getEndTime().isBefore(LocalDateTime.now())) {
    //         return Result.fail("秒杀已结束！");
    //     }
    //     // 库存小于1
    //     if (seckillVoucher.getStock() < 1) {
    //         return Result.fail("库存不足！");
    //     }
    //
    //     Long userId = UserHolder.getUser().getId();
    //
    //     // -- 实现悲观锁（只能用于单机
    //     /*
    //     * 为什么锁userId?
    //     *   因为我们只需要限制每个用户只能购买一次，所以这里锁的是userId，通过UserHolder可以得到当前用户
    //     * 为什么要进行.inter()？
    //     *   进入toString源码可以看到它每次都会new一个新的对象返回，会导致锁失效，而intern则会到常量池中找到相同字符的String返回
    //     * 为什么锁要包裹return？
    //     *   createVoucherOrder方法使用了事务，事务是整个方法包括方法内的return执行结束后spring才会进行提交事务
    //     *   如果在方法内对业务逻辑加锁，那么在释放锁时，事务还没提交，此时可能有其它线程得到锁执行逻辑 导致超卖
    //     * 为什么要获取proxy代理对象？
    //     *   在Java中，当一个对象调用它自己的方法时，该方法的调用通常会直接在当前对象上执行，而不会通过代理对象
    //     *   而事务的工作原理通常是通过代理对象来实现，这就导致事务可能会失效，所以我们需要手动通过当前对象的代理对象执行此方法
    //     *   这里手动获取代理对象的过程中，使用到了aspectj，需要引入pom依赖，在启动类加上注解 @EnableAspectJAutoProxy(exposeProxy = true)
    //     *   并且还要在IVoucherOrderService中声明createVoucherOrder方法
    //      * */
    //     // synchronized (userId.toString().intern()) {
    //     //     IVoucherOrderService proxy = (IVoucherOrderService) AopContext.currentProxy();
    //     //     return proxy.createVoucherOrder(voucherId);
    //     // }
    //
    //     // -- 实现Redis分布式锁（可用于集群
    //     // 针对用户上锁
    //     // SimpleRedisLock simpleRedisLock = new SimpleRedisLock("order:" + userId, stringRedisTemplate);
    //     // boolean lock = simpleRedisLock.tryLock(10L);
    //
    //     RLock rLock = redissonClient.getLock("order:" + userId);
    //     boolean lock = rLock.tryLock();
    //
    //     // 上锁失败 说明已被上锁
    //     if (!lock) {
    //         return Result.fail("一个用户只能购买一次，请勿重复购买！");
    //     }
    //
    //     try {
    //         IVoucherOrderService proxy = (IVoucherOrderService) AopContext.currentProxy();
    //         return proxy.createVoucherOrder(voucherId);
    //     } catch (IllegalStateException e) {
    //         throw new RuntimeException(e);
    //     } finally {
    //         // 解锁 业务代码有可能出现错误 所以这里放finally，无论业务是否成功 最后都会解锁
    //         // simpleRedisLock.unlock();
    //         rLock.unlock();
    //     }
    // }

    // 旧代码
    @Transactional
    public Result createVoucherOrder(Long voucherId) {
        Long userId = UserHolder.getUser().getId();

        Integer count = query().eq("voucher_id", voucherId).eq("user_id", userId).count();
        if (count > 0) {
            return Result.fail("用户已购买过一次！");
        }

        // 扣减库存
        // boolean reduceStock = seckillVoucherService.update().setSql("stock = stock - 1").eq("voucher_id", voucherId).update();
        boolean reduceStock = seckillVoucherMapper.reduceStock(voucherId);

        // 扣减库存失败
        if (!reduceStock) {
            return Result.fail("库存不足！");
        }

        // 创建订单
        VoucherOrder voucherOrder = new VoucherOrder();
        voucherOrder.setVoucherId(voucherId);
        long orderId = redisIdWorker.nextId("order");
        voucherOrder.setId(orderId);
        voucherOrder.setUserId(UserHolder.getUser().getId());
        voucherOrderService.save(voucherOrder);
        // 返回订单ID
        return Result.ok(orderId);
    }

    @Override
    @Transactional
    public void createVoucherOrder(VoucherOrder voucherOrder) {
        // 扣减库存
        // boolean reduceStock = seckillVoucherService.update().setSql("stock = stock - 1").eq("voucher_id", voucherId).update();
        boolean reduceStock = seckillVoucherMapper.reduceStock(voucherOrder.getVoucherId());

        // 扣减库存失败
        if (!reduceStock) {
            log.error("库存不足！");
            return;
        }
        // 成功 订单存入数据库
        save(voucherOrder);
    }
}
